#include "global.h"
#include "dp_network.h"




bool volatile abort_loop = 0;

unsigned int mysqld_port = 3358, select_errors, test_flags;
unsigned short mysqld_port_timeout = 10;
unsigned int back_log = 1024;
static char *my_bind_addr_str = NULL;
static int ip_sock;

void network_init(void)
{
    int arg;
    int ret;
    uint waited;
    uint this_wait;
    uint retry;
    char port_buf[NI_MAXSERV];

    ret = 0;

    if (mysqld_port != 0)
    {
        struct addrinfo *ai, *a;
        struct addrinfo hints;
        int error;
        dbug_print("IP Socket is %d\n", mysqld_port);

        bzero(&hints, sizeof (hints));
        hints.ai_flags = AI_PASSIVE;
        hints.ai_socktype = SOCK_STREAM;
        hints.ai_family = AF_UNSPEC;

        snprintf(port_buf, NI_MAXSERV, "%d", mysqld_port);
        error = getaddrinfo(my_bind_addr_str, port_buf, &hints, &ai);
        if (error != 0)
        {
            printf("error Got error: %d from getaddrinfo()", error);
            perror((char *) (ER_IPSOCK_ERROR - ER_ERROR_FIRST)); /* purecov: tested */
            exit(1); /* purecov: tested */
        }

        for (a = ai; a != NULL; a = a->ai_next)
        {
            ip_sock = socket(a->ai_family, a->ai_socktype, a->ai_protocol);
            if (ip_sock != INVALID_SOCKET)
                break;
        }

        if (ip_sock == INVALID_SOCKET)
        {
            printf("error Got error: %d from socket()", socket_errno);
            perror((char *) (ER_IPSOCK_ERROR - ER_ERROR_FIRST)); /* purecov: tested */
            exit(1); /* purecov: tested */
        }

        arg = 1;
        (void) setsockopt(ip_sock, SOL_SOCKET, SO_REUSEADDR, (char*) &arg, sizeof (arg));

        /*
          Sometimes the port is not released fast enough when stopping and
          restarting the server. This happens quite often with the test suite
          on busy Linux systems. Retry to bind the address at these intervals:
          Sleep intervals: 1, 2, 4,  6,  9, 13, 17, 22, ...
          Retry at second: 1, 3, 7, 13, 22, 35, 52, 74, ...
          Limit the sequence by mysqld_port_timeout (set --port-open-timeout=#).
         */
        for (waited = 0, retry = 1;; retry++, waited += this_wait)
        {
            if (((ret = bind(ip_sock, a->ai_addr, a->ai_addrlen)) >= 0) ||
                    (socket_errno != SOCKET_EADDRINUSE) ||
                    (waited >= mysqld_port_timeout))
                break;
            printf("Retrying bind on TCP/IP port %u", mysqld_port);
            this_wait = retry * retry / 3 + 1;
            sleep(this_wait);
        }
        freeaddrinfo(ai);
        if (ret < 0)
        {
            printf("error Got error: %d from bind", socket_errno);
            perror("Can't start server: Bind on TCP/IP port");
            printf("Do you already have another mysqld server running on port: %d ?", mysqld_port);
            exit(1);
        }
        if (listen(ip_sock, (int) back_log) < 0)
        {
            perror("Can't start server: listen() on TCP/IP port");
            printf("listen() on TCP/IP failed with error %d",
                    socket_errno);
            exit(1);
        }
    }

    dbug_print("server started");
    return;
}

void handle_connections_sockets(void)
{
    int sock = 0, new_sock = 0;
    uint retry;
    unsigned short error_count = 0;
    THD *thd;
    struct sockaddr_storage cAddr;
    int ip_flags = 0, socket_flags = 0, flags = 0, retval;
    fd_set readFDs, clientFDs;
    VIO *vio_tmp;
    unsigned int max_used_connection = (unsigned int) (ip_sock + 1);
    dbug_print("handle_connections_sockets");
    FD_ZERO(&clientFDs);

    if (ip_sock != INVALID_SOCKET)
    {
        FD_SET(ip_sock, &clientFDs);

#ifdef HAVE_FCNTL
        ip_flags = fcntl(ip_sock, F_GETFL, 0);
#endif

    }

    dbug_print("general Waiting for connections.");

    while (!abort_loop)
    {
        readFDs = clientFDs;
        retval = select((int) max_used_connection, &readFDs, 0, 0, 0);

        if (retval < 0)
        {
            if (socket_errno != SOCKET_EINTR)
            {
                if (!select_errors++ && !abort_loop) /* purecov: inspected */
                    printf("mysqld: Got error %d from select", socket_errno); /* purecov: inspected */
            }
            continue;
        }

        if (abort_loop)
        {
            break;
        }

        /* Is this a new connection request ? */
        sock = ip_sock;
        flags = ip_flags;

        if (!(test_flags & TEST_BLOCKING))
        {
#if defined(O_NONBLOCK)
            fcntl(sock, F_SETFL, flags | O_NONBLOCK);
#elif defined(O_NDELAY)
            fcntl(sock, F_SETFL, flags | O_NDELAY);
#endif
        }


        for (retry = 0; retry < MAX_ACCEPT_RETRY; retry++)
        {
            socklen_t length = sizeof (struct sockaddr_storage);
            new_sock = accept(sock, (struct sockaddr *) (&cAddr),
                    &length);
            if (new_sock != INVALID_SOCKET ||
                    (socket_errno != SOCKET_EINTR && socket_errno != SOCKET_EAGAIN))
                break;

            if (!(test_flags & TEST_BLOCKING))
            {
                if (retry == MAX_ACCEPT_RETRY - 1)
                    fcntl(sock, F_SETFL, flags); // Try without O_NONBLOCK
            }
        }

        if (!(test_flags & TEST_BLOCKING))
            fcntl(sock, F_SETFL, flags);

        if (new_sock == INVALID_SOCKET)
        {
            if ((error_count++ & 255) == 0) // This can happen often
                perror("Error in accept");

            if (socket_errno == SOCKET_ENFILE || socket_errno == SOCKET_EMFILE)
                sleep(1); // Give other threads some time
            continue;
        }

        socklen_t dummyLen;
        struct sockaddr_storage dummy;
        dummyLen = sizeof (dummy);
        if (getsockname(new_sock, (struct sockaddr *) &dummy,
                (socklen_t *) & dummyLen) < 0)
        {
            perror("Error on new connection socket");
            (void) shutdown(new_sock, SHUT_RDWR);
            (void) closesocket(new_sock);
            continue;
        }

        /*
         ** Don't allow too many connections
         */

        thd = (THD *) malloc(sizeof (THD));
        if (NULL == thd)
        {
            (void) shutdown(new_sock, SHUT_RDWR);
            (void) closesocket(new_sock);
            continue;
        }
        memset(thd, 0, sizeof (THD));

        if (!(vio_tmp = vio_new(new_sock, VIO_TYPE_TCPIP, 0))
                || my_net_init(thd->net, vio_tmp))
        {
            /*
              Only delete the temporary vio if we didn't already attach it to the
              NET object. The destructor in THD will delete any initialized net
              structure.
             */
            if (vio_tmp && thd->net != vio_tmp)
                free(vio_tmp);
            else
            {
                (void) shutdown(new_sock, SHUT_RDWR);
                (void) closesocket(new_sock);
            }
            free(thd);
            continue;
        }

        dbug_print("create_new_thread(thd)");
    }
    return;
}

VIO* vio_new(int sd, enum enum_vio_type type, uint flags)
{
    VIO *vio;
    dbug_print("vio_new");
    dbug_print("enter sd: %d", sd);

    vio = (VIO*) malloc(sizeof (VIO));
    if (vio)
    {
        bzero((char*) vio, sizeof (*vio));
        vio->type = type;
        vio->sd = sd;

        sprintf(vio->desc,
                (vio->type == VIO_TYPE_SOCKET ? "socket (%d)" : "TCP/IP (%d)"),
                vio->sd);
        /*
          We call fcntl() to set the flags and then immediately read them back
          to make sure that we and the system are in agreement on the state of
          things.

          An example of why we need to do this is FreeBSD (and apparently some
          other BSD-derived systems, like Mac OS X), where the system sometimes
          reports that the socket is set for non-blocking when it really will
          block.
         */
        fcntl(sd, F_SETFL, 0);
        vio->fcntl_mode = fcntl(sd, F_GETFL);
    }
    return vio;
}

bool my_net_init(VIO *net, VIO* vio)
{
    dbug_print("my_net_init");
    net = vio;
    return 0;
}


#if 0   // SQL CACHE

/**
  Send a single memory block from the query cache.

  Respects the client/server protocol limits for the
  size of the network packet, and splits a large block
  in pieces to ensure that individual piece doesn't exceed
  the maximal allowed size of the network packet (16M).

  @param[in] net NET handler
  @param[in] packet packet to send
  @param[in] len packet length

  @return Operation status
    @retval 0 On success
    @retval 1 On error
 */
static bool
send_data_in_chunks(NET *net, const uchar *packet, ulong len)
{
    /*
      On the client we may require more memory than max_allowed_packet
      to keep, both, the truncated last logical packet, and the
      compressed next packet.  This never (or in practice never)
      happens without compression, since without compression it's very
      unlikely that a) a truncated logical packet would remain on the
      client when it's time to read the next packet b) a subsequent
      logical packet that is being read would be so large that
      size-of-new-packet + size-of-old-packet-tail >
      max_allowed_packet.  To remedy this issue, we send data in 1MB
      sized packets, that's below the current client default of 16MB
      for max_allowed_packet, but large enough to ensure there is no
      unnecessary overhead from too many syscalls per result set.
     */
    static const ulong MAX_CHUNK_LENGTH = 1024 * 1024;

    while (len > MAX_CHUNK_LENGTH)
    {
        if (net_real_write(net, packet, MAX_CHUNK_LENGTH))
            return TRUE;
        packet += MAX_CHUNK_LENGTH;
        len -= MAX_CHUNK_LENGTH;
    }
    if (len && net_real_write(net, packet, len))
        return TRUE;

    return FALSE;
}

int
net_real_write(NET *net, const uchar *packet, size_t len)
{
    size_t length;
    const uchar *pos, *end;
    thr_alarm_t alarmed;
#ifndef NO_ALARM
    ALARM alarm_buff;
#endif
    uint retry_count = 0;
    my_bool net_blocking = vio_is_blocking(net->vio);
    DBUG_ENTER("net_real_write");

#if defined(MYSQL_SERVER) && defined(USE_QUERY_CACHE)
    query_cache_insert((char*) packet, len, net->pkt_nr);
#endif

    if (net->error == 2)
        DBUG_RETURN(-1); /* socket can't be used */

    net->reading_or_writing = 2;
#ifdef HAVE_COMPRESS
    if (net->compress)
    {
        size_t complen;
        uchar *b;
        uint header_length = NET_HEADER_SIZE + COMP_HEADER_SIZE;
        if (!(b = (uchar*) my_malloc(len + NET_HEADER_SIZE +
                COMP_HEADER_SIZE, MYF(MY_WME))))
        {
            net->error = 2;
            net->last_errno = ER_OUT_OF_RESOURCES;
            /* In the server, the error is reported by MY_WME flag. */
            net->reading_or_writing = 0;
            DBUG_RETURN(1);
        }
        memcpy(b + header_length, packet, len);

        if (my_compress(b + header_length, &len, &complen))
            complen = 0;
        int3store(&b[NET_HEADER_SIZE], complen);
        int3store(b, len);
        b[3] = (uchar) (net->compress_pkt_nr++);
        len += header_length;
        packet = b;
    }
#endif /* HAVE_COMPRESS */

#ifdef DEBUG_DATA_PACKETS
    DBUG_DUMP("data", packet, len);
#endif

#ifndef NO_ALARM
    thr_alarm_init(&alarmed);
    if (net_blocking)
        thr_alarm(&alarmed, net->write_timeout, &alarm_buff);
#else
    alarmed = 0;
    /* Write timeout is set in my_net_set_write_timeout */
#endif /* NO_ALARM */

    pos = packet;
    end = pos + len;
    while (pos != end)
    {
        if ((long) (length = vio_write(net->vio, pos, (size_t) (end - pos))) <= 0)
        {
            my_bool interrupted = vio_should_retry(net->vio);
#if !defined(__WIN__)
            if ((interrupted || length == 0) && !thr_alarm_in_use(&alarmed))
            {
                if (!thr_alarm(&alarmed, net->write_timeout, &alarm_buff))
                { /* Always true for client */
                    my_bool old_mode;
                    while (vio_blocking(net->vio, TRUE, &old_mode) < 0)
                    {
                        if (vio_should_retry(net->vio) && retry_count++ < net->retry_count)
                            continue;
#ifdef EXTRA_DEBUG
                        fprintf(stderr,
                                "%s: my_net_write: fcntl returned error %d, aborting thread\n",
                                my_progname, vio_errno(net->vio));
#endif /* EXTRA_DEBUG */
                        net->error = 2; /* Close socket */
                        net->last_errno = ER_NET_PACKET_TOO_LARGE;
#ifdef MYSQL_SERVER
                        my_error(ER_NET_PACKET_TOO_LARGE, MYF(0));
#endif
                        goto end;
                    }
                    retry_count = 0;
                    continue;
                }
            } else
#endif /* !defined(__WIN__) */
                if (thr_alarm_in_use(&alarmed) && !thr_got_alarm(&alarmed) &&
                    interrupted)
            {
                if (retry_count++ < net->retry_count)
                    continue;
#ifdef EXTRA_DEBUG
                fprintf(stderr, "%s: write looped, aborting thread\n",
                        my_progname);
#endif /* EXTRA_DEBUG */
            }
#if defined(THREAD_SAFE_CLIENT) && !defined(MYSQL_SERVER)
            if (vio_errno(net->vio) == SOCKET_EINTR)
            {
                DBUG_PRINT("warning", ("Interrupted write. Retrying..."));
                continue;
            }
#endif /* defined(THREAD_SAFE_CLIENT) && !defined(MYSQL_SERVER) */
            net->error = 2; /* Close socket */
            net->last_errno = (interrupted ? ER_NET_WRITE_INTERRUPTED :
                    ER_NET_ERROR_ON_WRITE);
#ifdef MYSQL_SERVER
            my_error(net->last_errno, MYF(0));
#endif /* MYSQL_SERVER */
            break;
        }
        pos += length;
        update_statistics(thd_increment_bytes_sent(length));
    }
#ifndef __WIN__
end:
#endif
#ifdef HAVE_COMPRESS
    if (net->compress)
        my_free((void*) packet);
#endif
    if (thr_alarm_in_use(&alarmed))
    {
        my_bool old_mode;
        thr_end_alarm(&alarmed);
        vio_blocking(net->vio, net_blocking, &old_mode);
    }
    net->reading_or_writing = 0;
    DBUG_RETURN(((int) (pos != end)));
}
#endif

static my_bool
net_write_buff(NET *net, const uchar *packet, ulong len)
{
    ulong left_length;
    if (net->compress && net->max_packet > MAX_PACKET_LENGTH)
        left_length = (ulong) (MAX_PACKET_LENGTH - (net->write_pos - net->buff));
    else
        left_length = (ulong) (net->buff_end - net->write_pos);

#ifdef DEBUG_DATA_PACKETS
    DBUG_DUMP("data", packet, len);
#endif
    if (len > left_length)
    {
        if (net->write_pos != net->buff)
        {
            /* Fill up already used packet and write it */
            memcpy((char*) net->write_pos, packet, left_length);
            if (net_real_write(net, net->buff,
                    (size_t) (net->write_pos - net->buff) + left_length))
                return 1;
            net->write_pos = net->buff;
            packet += left_length;
            len -= left_length;
        }
        if (net->compress)
        {
            /*
              We can't have bigger packets than 16M with compression
              Because the uncompressed length is stored in 3 bytes
             */
            left_length = MAX_PACKET_LENGTH;
            while (len > left_length)
            {
                if (net_real_write(net, packet, left_length))
                    return 1;
                packet += left_length;
                len -= left_length;
            }
        }
        if (len > net->max_packet)
            return net_real_write(net, packet, len) ? 1 : 0;
        /* Send out rest of the blocks as full sized blocks */
    }
    memcpy((char*) net->write_pos, packet, len);
    net->write_pos += len;
    return 0;
}

/*****************************************************************************
** Write something to server/client buffer
*****************************************************************************/

/**
  Write a logical packet with packet header.

  Format: Packet length (3 bytes), packet number(1 byte)
  When compression is used a 3 byte compression length is added

  @note
    If compression is used the original package is modified!
*/

my_bool
my_net_write(NET *net,const uchar *packet,size_t len)
{
  uchar buff[NET_HEADER_SIZE];
  int rc;

  if (unlikely(!net->vio)) /* nowhere to write */
    return 0;

  MYSQL_NET_WRITE_START(len);

  /*
    Big packets are handled by splitting them in packets of MAX_PACKET_LENGTH
    length. The last packet is always a packet that is < MAX_PACKET_LENGTH.
    (The last packet may even have a length of 0)
  */
  while (len >= MAX_PACKET_LENGTH)
  {
    const ulong z_size = MAX_PACKET_LENGTH;
    int3store(buff, z_size);
    buff[3]= (uchar) net->pkt_nr++;
    if (net_write_buff(net, buff, NET_HEADER_SIZE) ||
	net_write_buff(net, packet, z_size))
    {
      MYSQL_NET_WRITE_DONE(1);
      return 1;
    }
    packet += z_size;
    len-=     z_size;
  }
  /* Write last packet */
  int3store(buff,len);
  buff[3]= (uchar) net->pkt_nr++;
  if (net_write_buff(net, buff, NET_HEADER_SIZE))
  {
    MYSQL_NET_WRITE_DONE(1);
    return 1;
  }
#ifndef DEBUG_DATA_PACKETS
  DBUG_DUMP("packet_header", buff, NET_HEADER_SIZE);
#endif
  rc= test(net_write_buff(net,packet,len));
  MYSQL_NET_WRITE_DONE(rc);
  return rc;
}

/** Flush write_buffer if not empty. */

my_bool net_flush(NET *net)
{
    my_bool error = 0;
    dbug_print("net_flush");
    if (net->buff != net->write_pos)
    {
        error = test(net_real_write(net, net->buff,
                (size_t) (net->write_pos - net->buff)));
        net->write_pos = net->buff;
    }
    /* Sync packet number if using compression */
    if (net->compress)
        net->pkt_nr = net->compress_pkt_nr;
    DBUG_RETURN(error);
}